package com.ds.infrastructure.hyperspace.container.flow.handler.zone;

import com.ds.infrastructure.hyperspace.common.Container;
import com.ds.infrastructure.hyperspace.common.flow.FlowHandlerPipeline;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.handler.AbstractPreFlowHandler;
import com.ds.infrastructure.hyperspace.container.internal.operator.OperatorDefinition;
import com.ds.infrastructure.zonesdk.Zone;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.util.List;

/**
 * usage eg:
 *  -   handler: dynamicZonePreHandler
 *      handlerParams: [group, bj, uri|headerName, (operatorName,operatorArgs)...]
 * @author  Weishaoying
 */
@Slf4j
public class DynamicZonePreFlowHandler extends AbstractPreFlowHandler {

    private static final int INPUT_NAME_INDEX = 2;

    private volatile String shardingType;

    private volatile String defaultZone;

    private volatile String inputName;

    private volatile List<OperatorDefinition> operatorDefinitionList;

    @Override
    public void handleInput(HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        if (data == duplexFlowContext.getRequestStartState()) {
            if (inputName == null) {
                return;
            }
            String shardingKey = extractAndComputeInput(duplexFlowContext, inputName, operatorDefinitionList);
            Zone zone = findZone(duplexFlowContext, shardingType, shardingKey);
            assertZoneIsActive(defaultZone, zone, duplexFlowContext, shardingType, shardingKey);
        }
    }

    @Override
    public String handlerName() {
        return "dynamicZonePreHandler";
    }

    @Override
    public void initHandlerParameters(Container container, FlowHandlerPipeline pipeline, String[] handlerParams) {
        if (!StringUtils.isEmpty(handlerParams[0])) {
            shardingType = handlerParams[0];
        }
        if (handlerParams.length > 1 && !StringUtils.isEmpty(handlerParams[1])) {
            defaultZone = handlerParams[1];
        }
        if (handlerParams.length > INPUT_NAME_INDEX) {
            inputName = handlerParams[INPUT_NAME_INDEX];
            operatorDefinitionList = parseOperatorDefinition(container, handlerParams, INPUT_NAME_INDEX + 1);
        }
    }

}
